//! # Subscriber Trait 完整单元测试
//!
//! 提供 Subscriber trait 所有方法的全面测试覆盖

#[cfg(test)]
mod subscriber_tests {
    use crate::prelude::*;
    use crate::mock::{MockSubscriber, MockBehavior, test_helpers::*};
    use serde_json::json;
    use std::sync::Arc;
    use tokio::time::{timeout, Duration};
    use uuid::Uuid;

    /// 创建测试用的MockSubscriber
    async fn create_test_subscriber(pattern: &str) -> MockSubscriber {
        MockSubscriber::new(pattern)
    }

    /// 测试 Subscriber::try_receive 方法的所有场景
    mod try_receive_tests {
        use super::*;

        #[tokio::test]
        async fn test_try_receive_no_messages() {
            let subscriber = create_test_subscriber("test.topic").await;
            let result = subscriber.try_receive().await.unwrap();
            assert!(result.is_none());
        }

        #[tokio::test]
        async fn test_try_receive_with_messages() {
            let mut subscriber = create_test_subscriber("test.topic").await;
            
            // 添加测试消息
            let message = create_test_message("test.topic", json!({"data": "test"}));
            subscriber.add_message(message.clone()).await;
            
            let result = subscriber.try_receive().await.unwrap();
            assert!(result.is_some());
            
            let received = result.unwrap();
            assert_eq!(received.topic(), "test.topic");
            assert_eq!(received.payload()["data"], "test");
        }

        #[tokio::test]
        async fn test_try_receive_multiple_messages() {
            let mut subscriber = create_test_subscriber("test.topic").await;
            
            // 添加多个消息
            for i in 0..5 {
                let message = create_test_message(
                    "test.topic",
                    json!({"id": i, "data": format!("message_{}", i)})
                );
                subscriber.add_message(message).await;
            }
            
            // 依次接收消息
            for i in 0..5 {
                let result = subscriber.try_receive().await.unwrap();
                assert!(result.is_some());
                
                let received = result.unwrap();
                assert_eq!(received.payload()["id"], i);
            }
            
            // 所有消息接收完毕后应该返回None
            let result = subscriber.try_receive().await.unwrap();
            assert!(result.is_none());
        }

        #[tokio::test]
        async fn test_try_receive_after_close() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.close().await.unwrap();
            
            let result = subscriber.try_receive().await;
            assert!(result.is_err());
            assert!(result.unwrap_err().to_string().contains("订阅者已关闭"));
        }

        #[tokio::test]
        async fn test_try_receive_failure() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.set_behavior("try_receive", MockBehavior::new().with_failure("接收失败")).await;
            
            let result = subscriber.try_receive().await;
            assert!(result.is_err());
            assert!(result.unwrap_err().to_string().contains("接收失败"));
        }

        #[tokio::test]
        async fn test_try_receive_with_delay() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.set_behavior("try_receive", MockBehavior::new().with_delay(50)).await;
            
            let start = std::time::Instant::now();
            let result = subscriber.try_receive().await;
            
            assert!(result.is_ok());
            assert!(start.elapsed().as_millis() >= 50);
        }
    }

    /// 测试 Subscriber::receive 方法的所有场景
    mod receive_tests {
        use super::*;

        #[tokio::test]
        async fn test_receive_immediate_message() {
            let mut subscriber = create_test_subscriber("test.topic").await;
            
            let message = create_test_message("test.topic", json!({"immediate": true}));
            subscriber.add_message(message.clone()).await;
            
            let result = subscriber.receive().await.unwrap();
            assert!(result.is_some());
            assert_eq!(result.unwrap().payload()["immediate"], true);
        }

        #[tokio::test]
        async fn test_receive_no_message() {
            let subscriber = create_test_subscriber("test.topic").await;
            
            // 在Mock实现中，没有消息时应该返回None
            let result = subscriber.receive().await.unwrap();
            assert!(result.is_none());
        }

        #[tokio::test]
        async fn test_receive_different_message_types() {
            let mut subscriber = create_test_subscriber("test.topic").await;
            
            // 添加不同类型的消息
            let event = create_event_message("test.topic", json!({"type": "event"}));
            let command = create_command_message("test.topic", json!({"type": "command"}));
            let query = create_request_message("test.topic", json!({"type": "query"}), "reply.topic");
            
            subscriber.add_message(event).await;
            subscriber.add_message(command).await;
            subscriber.add_message(query).await;
            
            // 依次接收不同类型的消息
            let received_event = subscriber.receive().await.unwrap().unwrap();
            assert_eq!(received_event.metadata.attributes.message_type, MessageType::Event);
            
            let received_command = subscriber.receive().await.unwrap().unwrap();
            assert_eq!(received_command.metadata.attributes.message_type, MessageType::Command);
            
            let received_query = subscriber.receive().await.unwrap().unwrap();
            assert_eq!(received_query.metadata.attributes.message_type, MessageType::Query);
        }

        #[tokio::test]
        async fn test_receive_with_correlation_id() {
            let mut subscriber = create_test_subscriber("test.topic").await;
            let correlation_id = Uuid::new_v4();
            
            let message = create_correlated_message(
                "test.topic",
                json!({"correlated": true}),
                correlation_id
            );
            subscriber.add_message(message).await;
            
            let result = subscriber.receive().await.unwrap().unwrap();
            assert_eq!(result.metadata.correlation_id, Some(correlation_id));
        }

        #[tokio::test]
        async fn test_receive_after_close() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.close().await.unwrap();
            
            let result = subscriber.receive().await;
            assert!(result.is_err());
        }

        #[tokio::test]
        async fn test_receive_failure() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.set_behavior("receive", MockBehavior::new().with_failure("阻塞接收失败")).await;
            
            let result = subscriber.receive().await;
            assert!(result.is_err());
        }
    }

    /// 测试 Subscriber::receive_timeout 方法的所有场景
    mod receive_timeout_tests {
        use super::*;

        #[tokio::test]
        async fn test_receive_timeout_immediate_message() {
            let mut subscriber = create_test_subscriber("test.topic").await;
            
            let message = create_test_message("test.topic", json!({"timeout_test": true}));
            subscriber.add_message(message).await;
            
            let result = subscriber.receive_timeout(1000).await.unwrap();
            assert!(result.is_some());
        }

        #[tokio::test]
        async fn test_receive_timeout_no_message() {
            let subscriber = create_test_subscriber("test.topic").await;
            
            let start = std::time::Instant::now();
            let result = subscriber.receive_timeout(100).await.unwrap();
            let elapsed = start.elapsed();
            
            assert!(result.is_none());
            assert!(elapsed.as_millis() >= 100);
        }

        #[tokio::test]
        async fn test_receive_timeout_zero_timeout() {
            let subscriber = create_test_subscriber("test.topic").await;
            
            let result = subscriber.receive_timeout(0).await.unwrap();
            assert!(result.is_none());
        }

        #[tokio::test]
        async fn test_receive_timeout_large_timeout() {
            let mut subscriber = create_test_subscriber("test.topic").await;
            
            // 在短时间后添加消息，但设置长超时
            let message = create_test_message("test.topic", json!({"late": true}));
            subscriber.add_message(message).await;
            
            let result = subscriber.receive_timeout(5000).await.unwrap();
            assert!(result.is_some());
        }

        #[tokio::test]
        async fn test_receive_timeout_different_durations() {
            let subscriber = create_test_subscriber("test.topic").await;
            
            let timeouts = vec![10, 50, 100, 500];
            
            for timeout_ms in timeouts {
                let start = std::time::Instant::now();
                let result = subscriber.receive_timeout(timeout_ms).await.unwrap();
                let elapsed = start.elapsed();
                
                assert!(result.is_none());
                assert!(elapsed.as_millis() >= timeout_ms as u128);
            }
        }

        #[tokio::test]
        async fn test_receive_timeout_after_close() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.close().await.unwrap();
            
            let result = subscriber.receive_timeout(1000).await;
            assert!(result.is_err());
        }

        #[tokio::test]
        async fn test_receive_timeout_failure() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.set_behavior("receive_timeout", MockBehavior::new().with_failure("超时接收失败")).await;
            
            let result = subscriber.receive_timeout(1000).await;
            assert!(result.is_err());
        }
    }

    /// 测试 Subscriber::ack 方法的所有场景
    mod ack_tests {
        use super::*;

        #[tokio::test]
        async fn test_ack_single_message() {
            let subscriber = create_test_subscriber("test.topic").await;
            let message = create_test_message("test.topic", json!({"ack_test": true}));
            
            let result = subscriber.ack(&message).await;
            assert!(result.is_ok());
            
            let acked = subscriber.get_acknowledged_messages();
            assert_eq!(acked.len(), 1);
            assert_eq!(acked[0], message.id());
        }

        #[tokio::test]
        async fn test_ack_multiple_messages() {
            let subscriber = create_test_subscriber("test.topic").await;
            
            let mut message_ids = Vec::new();
            for i in 0..5 {
                let message = create_test_message(
                    "test.topic",
                    json!({"id": i})
                );
                message_ids.push(message.id());
                subscriber.ack(&message).await.unwrap();
            }
            
            let acked = subscriber.get_acknowledged_messages();
            assert_eq!(acked.len(), 5);
            
            for id in message_ids {
                assert!(acked.contains(&id));
            }
        }

        #[tokio::test]
        async fn test_ack_same_message_multiple_times() {
            let subscriber = create_test_subscriber("test.topic").await;
            let message = create_test_message("test.topic", json!({"duplicate": true}));
            
            // 多次确认同一条消息
            subscriber.ack(&message).await.unwrap();
            subscriber.ack(&message).await.unwrap();
            subscriber.ack(&message).await.unwrap();
            
            let acked = subscriber.get_acknowledged_messages();
            assert_eq!(acked.len(), 3); // Mock允许重复确认
        }

        #[tokio::test]
        async fn test_ack_different_message_types() {
            let subscriber = create_test_subscriber("test.topic").await;
            
            let event = create_event_message("test.topic", json!({"type": "event"}));
            let command = create_command_message("test.topic", json!({"type": "command"}));
            let query = create_request_message("test.topic", json!({"type": "query"}), "reply");
            
            subscriber.ack(&event).await.unwrap();
            subscriber.ack(&command).await.unwrap();
            subscriber.ack(&query).await.unwrap();
            
            let acked = subscriber.get_acknowledged_messages();
            assert_eq!(acked.len(), 3);
        }

        #[tokio::test]
        async fn test_ack_after_close() {
            let subscriber = create_test_subscriber("test.topic").await;
            let message = create_test_message("test.topic", json!({"after_close": true}));
            
            subscriber.close().await.unwrap();
            
            let result = subscriber.ack(&message).await;
            assert!(result.is_err());
        }

        #[tokio::test]
        async fn test_ack_failure() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.set_behavior("ack", MockBehavior::new().with_failure("确认失败")).await;
            
            let message = create_test_message("test.topic", json!({"fail": true}));
            let result = subscriber.ack(&message).await;
            assert!(result.is_err());
        }
    }

    /// 测试 Subscriber::nack 方法的所有场景
    mod nack_tests {
        use super::*;

        #[tokio::test]
        async fn test_nack_without_requeue() {
            let subscriber = create_test_subscriber("test.topic").await;
            let message = create_test_message("test.topic", json!({"nack_test": true}));
            
            let result = subscriber.nack(&message, false).await;
            assert!(result.is_ok());
            
            let nacked = subscriber.get_nacked_messages();
            assert_eq!(nacked.len(), 1);
            assert_eq!(nacked[0].0, message.id());
            assert_eq!(nacked[0].1, false); // requeue = false
        }

        #[tokio::test]
        async fn test_nack_with_requeue() {
            let mut subscriber = create_test_subscriber("test.topic").await;
            let message = create_test_message("test.topic", json!({"requeue_test": true}));
            
            let result = subscriber.nack(&message, true).await;
            assert!(result.is_ok());
            
            let nacked = subscriber.get_nacked_messages();
            assert_eq!(nacked.len(), 1);
            assert_eq!(nacked[0].0, message.id());
            assert_eq!(nacked[0].1, true); // requeue = true
            
            // 消息应该被重新放入队列
            assert_eq!(subscriber.message_count(), 1);
        }

        #[tokio::test]
        async fn test_nack_multiple_messages() {
            let subscriber = create_test_subscriber("test.topic").await;
            
            let messages = create_batch_messages("test.topic", 3);
            
            // 不同的requeue策略
            subscriber.nack(&messages[0], false).await.unwrap();
            subscriber.nack(&messages[1], true).await.unwrap();
            subscriber.nack(&messages[2], false).await.unwrap();
            
            let nacked = subscriber.get_nacked_messages();
            assert_eq!(nacked.len(), 3);
            
            // 检查requeue策略
            assert_eq!(nacked[0].1, false);
            assert_eq!(nacked[1].1, true);
            assert_eq!(nacked[2].1, false);
        }

        #[tokio::test]
        async fn test_nack_same_message_multiple_times() {
            let subscriber = create_test_subscriber("test.topic").await;
            let message = create_test_message("test.topic", json!({"multi_nack": true}));
            
            subscriber.nack(&message, false).await.unwrap();
            subscriber.nack(&message, true).await.unwrap();
            
            let nacked = subscriber.get_nacked_messages();
            assert_eq!(nacked.len(), 2);
        }

        #[tokio::test]
        async fn test_nack_after_close() {
            let subscriber = create_test_subscriber("test.topic").await;
            let message = create_test_message("test.topic", json!({"after_close": true}));
            
            subscriber.close().await.unwrap();
            
            let result = subscriber.nack(&message, false).await;
            assert!(result.is_err());
        }

        #[tokio::test]
        async fn test_nack_failure() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.set_behavior("nack", MockBehavior::new().with_failure("拒绝失败")).await;
            
            let message = create_test_message("test.topic", json!({"fail": true}));
            let result = subscriber.nack(&message, false).await;
            assert!(result.is_err());
        }
    }

    /// 测试 Subscriber 的状态方法
    mod state_tests {
        use super::*;

        #[tokio::test]
        async fn test_subscription_id() {
            let subscriber = create_test_subscriber("test.topic").await;
            let id = subscriber.subscription_id();
            assert!(!id.is_empty());
        }

        #[tokio::test]
        async fn test_pattern() {
            let pattern = "test.pattern.*";
            let subscriber = create_test_subscriber(pattern).await;
            assert_eq!(subscriber.pattern(), pattern);
        }

        #[tokio::test]
        async fn test_is_active_initially() {
            let subscriber = create_test_subscriber("test.topic").await;
            assert!(subscriber.is_active());
        }

        #[tokio::test]
        async fn test_is_active_after_close() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.close().await.unwrap();
            assert!(!subscriber.is_active());
        }
    }

    /// 测试 Subscriber::close 方法的所有场景
    mod close_tests {
        use super::*;

        #[tokio::test]
        async fn test_close_clean_shutdown() {
            let mut subscriber = create_test_subscriber("test.topic").await;
            
            // 添加一些消息
            let message = create_test_message("test.topic", json!({"before_close": true}));
            subscriber.add_message(message).await;
            
            let result = subscriber.close().await;
            assert!(result.is_ok());
            assert!(!subscriber.is_active());
            
            // 关闭后消息队列应该被清空
            assert_eq!(subscriber.message_count(), 0);
        }

        #[tokio::test]
        async fn test_close_multiple_times() {
            let subscriber = create_test_subscriber("test.topic").await;
            
            let result1 = subscriber.close().await;
            let result2 = subscriber.close().await;
            
            assert!(result1.is_ok());
            assert!(result2.is_ok()); // 多次关闭应该成功
        }

        #[tokio::test]
        async fn test_close_failure() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.set_behavior("close", MockBehavior::new().with_failure("关闭失败")).await;
            
            let result = subscriber.close().await;
            assert!(result.is_err());
        }
    }

    /// 测试 Subscriber::stats 方法的所有场景
    mod stats_tests {
        use super::*;

        #[tokio::test]
        async fn test_stats_empty_subscriber() {
            let subscriber = create_test_subscriber("test.topic").await;
            let stats = subscriber.stats().await.unwrap();
            
            assert_eq!(stats.total_received, 0);
            assert_eq!(stats.total_acked, 0);
            assert_eq!(stats.total_nacked, 0);
            assert_eq!(stats.queue_length, 0);
        }

        #[tokio::test]
        async fn test_stats_with_activity() {
            let subscriber = create_test_subscriber("test.topic").await;
            
            // 模拟一些活动
            let message1 = create_test_message("test.topic", json!({"id": 1}));
            let message2 = create_test_message("test.topic", json!({"id": 2}));
            let message3 = create_test_message("test.topic", json!({"id": 3}));
            
            subscriber.ack(&message1).await.unwrap();
            subscriber.ack(&message2).await.unwrap();
            subscriber.nack(&message3, false).await.unwrap();
            
            let stats = subscriber.stats().await.unwrap();
            assert_eq!(stats.total_acked, 2);
            assert_eq!(stats.total_nacked, 1);
        }

        #[tokio::test]
        async fn test_stats_failure() {
            let subscriber = create_test_subscriber("test.topic").await;
            subscriber.set_behavior("stats", MockBehavior::new().with_failure("统计失败")).await;
            
            let result = subscriber.stats().await;
            assert!(result.is_err());
        }
    }

    /// 并发测试
    mod concurrent_tests {
        use super::*;
        use std::sync::atomic::{AtomicUsize, Ordering};

        #[tokio::test]
        async fn test_concurrent_receive() {
            let mut subscriber = Arc::new(create_test_subscriber("concurrent.test").await);
            
            // 添加足够的消息供并发接收
            for i in 0..20 {
                let message = create_test_message(
                    "concurrent.test",
                    json!({"id": i})
                );
                // 注意：由于需要可变引用，这里需要特殊处理
                let subscriber_mut = Arc::get_mut(&mut subscriber).unwrap();
                subscriber_mut.add_message(message).await;
            }
            
            let subscriber = Arc::new(subscriber.clone());
            let received_count = Arc::new(AtomicUsize::new(0));
            let mut handles = vec![];
            
            // 启动多个并发接收任务
            for _i in 0..5 {
                let sub_clone = subscriber.clone();
                let counter = received_count.clone();
                let handle = tokio::spawn(async move {
                    while let Ok(Some(_)) = sub_clone.try_receive().await {
                        counter.fetch_add(1, Ordering::Relaxed);
                    }
                });
                handles.push(handle);
            }
            
            // 等待所有任务完成
            for handle in handles {
                handle.await.unwrap();
            }
            
            // 验证所有消息都被接收
            assert_eq!(received_count.load(Ordering::Relaxed), 20);
        }

        #[tokio::test]
        async fn test_concurrent_ack_nack() {
            let subscriber = Arc::new(create_test_subscriber("concurrent.ack").await);
            let ack_count = Arc::new(AtomicUsize::new(0));
            let nack_count = Arc::new(AtomicUsize::new(0));
            
            let mut handles = vec![];
            
            // 并发确认消息
            for i in 0..10 {
                let sub_clone = subscriber.clone();
                let counter = ack_count.clone();
                let handle = tokio::spawn(async move {
                    let message = create_test_message(
                        "concurrent.ack",
                        json!({"ack_id": i})
                    );
                    if sub_clone.ack(&message).await.is_ok() {
                        counter.fetch_add(1, Ordering::Relaxed);
                    }
                });
                handles.push(handle);
            }
            
            // 并发拒绝消息
            for i in 0..5 {
                let sub_clone = subscriber.clone();
                let counter = nack_count.clone();
                let handle = tokio::spawn(async move {
                    let message = create_test_message(
                        "concurrent.ack",
                        json!({"nack_id": i})
                    );
                    if sub_clone.nack(&message, false).await.is_ok() {
                        counter.fetch_add(1, Ordering::Relaxed);
                    }
                });
                handles.push(handle);
            }
            
            // 等待所有操作完成
            for handle in handles {
                handle.await.unwrap();
            }
            
            assert_eq!(ack_count.load(Ordering::Relaxed), 10);
            assert_eq!(nack_count.load(Ordering::Relaxed), 5);
        }
    }

    /// 集成测试：测试完整的消息处理流程
    mod integration_tests {
        use super::*;

        #[tokio::test]
        async fn test_message_lifecycle() {
            let mut subscriber = create_test_subscriber("lifecycle.test").await;
            
            // 1. 添加消息
            let message = create_test_message(
                "lifecycle.test",
                json!({"lifecycle": "test"})
            );
            subscriber.add_message(message.clone()).await;
            
            // 2. 接收消息
            let received = subscriber.receive().await.unwrap();
            assert!(received.is_some());
            let received_msg = received.unwrap();
            
            // 3. 验证消息内容
            assert_eq!(received_msg.topic(), message.topic());
            assert_eq!(received_msg.payload(), message.payload());
            
            // 4. 确认消息
            subscriber.ack(&received_msg).await.unwrap();
            
            // 5. 验证确认记录
            let acked = subscriber.get_acknowledged_messages();
            assert_eq!(acked.len(), 1);
            assert_eq!(acked[0], received_msg.id());
        }

        #[tokio::test]
        async fn test_error_recovery_flow() {
            let mut subscriber = create_test_subscriber("error.recovery").await;
            
            // 添加消息
            let message = create_test_message(
                "error.recovery",
                json!({"retry": true})
            );
            subscriber.add_message(message.clone()).await;
            
            // 第一次处理失败，拒绝并重新入队
            let received1 = subscriber.receive().await.unwrap().unwrap();
            subscriber.nack(&received1, true).await.unwrap();
            
            // 消息应该重新可用
            let received2 = subscriber.receive().await.unwrap().unwrap();
            assert_eq!(received2.id(), received1.id());
            
            // 第二次处理成功
            subscriber.ack(&received2).await.unwrap();
            
            // 验证处理记录
            let nacked = subscriber.get_nacked_messages();
            let acked = subscriber.get_acknowledged_messages();
            
            assert_eq!(nacked.len(), 1);
            assert_eq!(acked.len(), 1);
            assert_eq!(nacked[0].1, true); // requeue = true
        }

        #[tokio::test]
        async fn test_timeout_handling() {
            let subscriber = create_test_subscriber("timeout.test").await;
            
            // 测试各种超时场景
            let short_timeout = subscriber.receive_timeout(10).await.unwrap();
            assert!(short_timeout.is_none());
            
            let medium_timeout = subscriber.receive_timeout(100).await.unwrap();
            assert!(medium_timeout.is_none());
            
            // 添加消息后应该立即接收到
            let mut subscriber_mut = subscriber;
            let message = create_test_message("timeout.test", json!({"immediate": true}));
            subscriber_mut.add_message(message).await;
            
            let immediate = subscriber_mut.receive_timeout(1000).await.unwrap();
            assert!(immediate.is_some());
        }
    }
}